package io.yunho.yunho.interview.two.java8.webflux;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;

import java.time.LocalDateTime;

/**
 * @author: Jason
 * @date: 2021/5/12 9:54
 * @description:
 */
public class MonoTest {
    public static void main(String[] args) {
        //reactor = jdk8 stream + jdk9 Reactive stream
        //mono 0-1个元素
        // flux 0-N个元素
        String[] strings = {"1","2","3"};
        //定义订阅者
        Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            private Subscription subscription;
            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                System.out.println("subscription***********"+subscription);
                this.subscription.request(1);
            }

            @Override
            public void onNext(Integer integer) {
                System.out.println("接收到数据："+integer);
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                throwable.printStackTrace();
                System.out.println("异常信息："+throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("处理完成："+ LocalDateTime.now());
            }
        };
        //这里是jdk8的stream
        Flux.fromArray(strings).map(s-> Integer.parseInt(s))
                //最终操作
                //这里就是jdk9的reactive stream
                .subscribe(subscriber);
    }
}
